package com.mygame.message.rpc;

import java.util.concurrent.atomic.AtomicInteger;

import com.mygame.common.codec.DataCodecKit;
import com.mygame.common.protobuf.ExternalMessage;
import io.nats.streaming.StreamingConnection;
//import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
//import org.springframework.kafka.core.KafkaTemplate;
import com.mygame.common.cloud.PlayerServiceInstance;
import com.mygame.common.utils.TopicUtil;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;

public class GameRpcService {

    private AtomicInteger seqId = new AtomicInteger();// 自增的唯一序列Id
    private int localServerId;// 本地服务实例ID

    private PlayerServiceInstance playerServiceInstance;
    private EventExecutorGroup eventExecutorGroup;
    private StreamingConnection sc;
    private static Logger logger = LoggerFactory.getLogger(GameRpcService.class);
    private GameRpcCallbackService gameRpcCallbackService;
    private String requestTopic;
    private String responseTopic;

    public GameRpcService(String requestTopic,String responseTopic,int localServerId, PlayerServiceInstance playerServiceInstance, EventExecutorGroup eventExecutorGroup, StreamingConnection sc) {
        this.localServerId = localServerId;
        this.requestTopic = requestTopic;
        this.responseTopic = responseTopic;
        this.playerServiceInstance = playerServiceInstance;
        this.eventExecutorGroup = eventExecutorGroup;
        this.sc = sc;
        this.gameRpcCallbackService = new GameRpcCallbackService(eventExecutorGroup);
    }
    public void sendRPCResponse(ExternalMessage gameMessage) {
        String sendTopic = TopicUtil.generateTopic(responseTopic, gameMessage.getToServerId());
        byte[] value = DataCodecKit.encode(gameMessage);
        try {
            sc.publish(sendTopic, value);
        }catch (Exception e){
            System.out.println(e);
        }
    }
    public void receiveRpcResponse(ExternalMessage gameMessage) {
        gameRpcCallbackService.callback(gameMessage);
    }

    public void sendRPCRequest(ExternalMessage gameMessage, Promise<ExternalMessage> promise) {
        long playerId = gameMessage.getPlayerId();
        int serviceId = gameMessage.getServiceId();

//        GameMessagePackage gameMessagePackage = new GameMessagePackage();
//        gameMessagePackage.setHeader(gameMessage.getHeader());
//        gameMessagePackage.setBody(gameMessage.body());
//        GameMessageHeader header = gameMessage.getHeader();
//        header.setClientSeqId(seqId.incrementAndGet());
//        header.setFromServerId(localServerId);
//        header.setClientSendTime(System.currentTimeMillis());
        gameMessage.setFromServerId(localServerId);
        gameMessage.setClientSeqId(seqId.incrementAndGet());
        Promise<Integer> selectServerIdPromise = new DefaultPromise<>(this.eventExecutorGroup.next());
        playerServiceInstance.selectServerId(playerId, serviceId, selectServerIdPromise).addListener(new GenericFutureListener<Future<Integer>>() {
            @Override
            public void operationComplete(Future<Integer> future) throws Exception {
                if (future.isSuccess()) {
                    gameMessage.setToServerId(future.get());
                    // 动态创建游戏网关监听消息的topic
                    String sendTopic = TopicUtil.generateTopic(requestTopic, gameMessage.getToServerId());
                    byte[] value = DataCodecKit.encode(gameMessage);
                    try {
                        sc.publish(sendTopic, value);
                    }catch (Exception e){
                        System.out.println(e);
                    }
                    gameRpcCallbackService.addCallback(gameMessage.getClientSeqId(), promise);
                } else {
                    logger.error("获取目标服务实例ID出错",future.cause());
                }
            }
        });
    }

}
